package org.atguigu.gmall.realtime.join;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.atguigu.gmall.realtime.app.BaseSQLApp;
import org.atguigu.gmall.realtime.util.SQLUtil;

import java.time.Duration;

/**
 * @Author zenghaifeng
 * @Date 2023/4/25 0:12
 * @Package: org.atguigu.gmall.realtime.join
 * @Class: LeftJoinDemo
 * @Description: TODO
 * @Version 1.0
 */
public class LeftJoinDemo extends BaseSQLApp {
    public static void main(String[] args) {
        new LeftJoinDemo().init(50000,2,"LeftJoinDemo");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tbe) {
        tbe.executeSql("create table a(" +
                " id string, " +
                " name string, " +
                ")" + SQLUtil.getKafkaSourceDDL("a", "LeftJoinDemo_1", "csv"));

        tbe.executeSql("create table b(" +
                " id string, " +
                " age int " +
                ")" + SQLUtil.getKafkaSourceDDL("b", "LeftJoinDemo_2", "csv"));

        tbe.getConfig().setIdleStateRetention(Duration.ofSeconds(20));

        tbe.sqlQuery("select " +
                        " a.id id," +
                        " name, " +
                        " age " +
                        "from a " +
                        "left join b on a.id=b.id")
                .execute()
                .print();
    }
}
